package RocketMQ;

import com.maomao.common.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.jupiter.api.Test;


import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @author mao
 * @date 2025-05-29- 14:51
 * @descript
 */
@Slf4j
public class RocketMqDemo {

    private static final String ROCKETMQ_SERVER = "192.100.4.19:9876";

    /**
     * 消费者
     * @author mao
     * @date 2025/5/29 14:59
     */
    @Test
    public void consumerTest() throws Exception{
        // 实例化消费者
        DefaultMQPushConsumer consumer = new
                DefaultMQPushConsumer("maomao_consumer_group");
        // 设置NameServer的地址
        consumer.setNamesrvAddr(ROCKETMQ_SERVER);
        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("maomao_topic", "*");
        // 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt>
                                                                    msgs, ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n",
                        Thread.currentThread().getName(), msgs);
                for (int i = 0; i < msgs.size(); i++) {
                    MessageExt msg = msgs.get(i);
                    String content = new String(msg.getBody());
                    log.info("收到消息：{}", msg.getMsgId() + " " + msg.getTopic()
                            + " " + msg.getTags() + " " + content);
                    try {
                    //消费者的业务代码
                    // redisSeckillServiceImpl.executeSeckill(dto);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
        ThreadUtil.sleepMilliSeconds(Integer.MAX_VALUE);
    }

    /**
     * 同步消息
     */
    @Test
    public void produceSynTest() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new
                DefaultMQProducer("maomao_producer_group");
        // 设置NameServer的地址
        producer.setNamesrvAddr(ROCKETMQ_SERVER);
        // 启动Producer实例
        producer.start();
        for (int i = 0; i < 100; i++) {
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("maomao_topic" /* Topic */,
                    "TagA" /* Tag */,
                    ("Hello RocketMQ " +
                            i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
            // 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            System.out.printf("%s%n", sendResult);
        }
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }

    /**
     * 异步消息
     */
    @Test
    public void produceAsynTest() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new
                DefaultMQProducer("maomao_producer_group");
        // 设置NameServer的地址
        producer.setNamesrvAddr(ROCKETMQ_SERVER);
        // 启动Producer实例
        producer.start();
        producer.setRetryTimesWhenSendAsyncFailed(0);
        int messageCount = 100;
        // 根据消息数量实例化倒计时计算器
        final CountDownLatch countDownLatch = new CountDownLatch(messageCount);
        for (int i = 0; i < messageCount; i++) {
            final int index = i;
            // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    ("Helloworld"+index).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    // SendCallback接收异步返回结果的回调
                    producer.send(msg, new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            System.out.printf("%-10d OK %s %n", index,
                                    sendResult.getMsgId());
                        }
                        @Override
                        public void onException(Throwable e) {
                            System.out.printf("%-10d Exception %s %n", index, e);
                            e.printStackTrace();
                        }
                    });
        }
        // 等待5s
        countDownLatch.await(5, TimeUnit.SECONDS);
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }

    /**
     * 单向消息
     * producer向 broker 发送消息，执行 API 时直接返回，不等待broker 服务器的结果 。
     */
    @Test
    public void onewayProducer() throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new
                DefaultMQProducer("please_rename_unique_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr(ROCKETMQ_SERVER);
        // 启动Producer实例
        producer.start();
        for (int i = 0; i < 100; i++) {
        // 创建消息，并指定Topic，Tag和消息体
            Message msg = new Message("TopicTest" /* Topic */,
                    "TagA" /* Tag */,
                    ("Java高并发 卷王" +
                            i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
            );
        // 发送单向消息，没有任何返回结果
            producer.sendOneway(msg);
        }
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }


}
